package com.wan37.gate.global.dispatcher;

import com.google.protobuf.Message;
import com.wan37.gate.component.MessageQueue;
import com.wan37.gate.domain.ClientConnection;
import com.wan37.gate.domain.RequestMessage;
import com.wan37.gate.global.maps.ClientConnectionMap;
import com.wan37.gate.handler.GateToAuthConnectionHandler;
import com.wan37.gate.handler.GateToLogicConnectionHandler;
import com.wan37.gate.utils.RouteUtils;
import com.wan37.protobuf.analysis.ParseMap;
import com.wan37.protobuf.constant.Common;
import com.wan37.protobuf.constant.PtoNum;
import com.wan37.protobuf.message.client2server.auth.Auth;
import com.wan37.protobuf.message.internal.Internal;
import com.wan37.protobuf.util.ServerProtoUtils;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.concurrent.*;

/**
 * @author : luoyong
 * @date : 2020-05-26 17:40
 **/
public final class ClientMessageDispatcher implements Runnable {

    private static final Logger logger = LoggerFactory.getLogger(ClientMessageDispatcher.class);

    /**
     * 分发器中的session缓存区, 用于缓存客户端连接及其对应的消息队列.
     * 每一个netId都对应一个MessageQueue.
     * key = netId, value = 该客户端连接的消息队列
     */
    private final Map<Long, MessageQueue> clientSessionMap;

    /**
     * 线程池(处理每个messageQueue)
     */
    private Executor handleMessageQueueExecutor;

    /**
     * 轮询周期
     */
    private long sleepTime;

    //构造函数
    public ClientMessageDispatcher() {
        this.clientSessionMap = new ConcurrentHashMap<>();
        this.handleMessageQueueExecutor = new ThreadPoolExecutor(10, 10, 50,
                TimeUnit.SECONDS, new LinkedBlockingQueue<>());
        //200毫秒
        this.sleepTime = 200L;
    }

    /**
     * 向分发器的缓存中添加客户端channel对应的netId 和 该客户端的messageQueue。
     *
     * @param netId
     * @param messageQueue
     */
    public void addMessageQueue(Long netId, MessageQueue messageQueue) {
        this.clientSessionMap.put(netId, messageQueue);
    }

    public void addMessageQueue(ClientConnection clientConnection, MessageQueue messageQueue) {
        this.clientSessionMap.put(clientConnection.getNetId(), messageQueue);
    }

    /**
     * 返回netId对应的客户端的messageQueue。
     *
     * @param netId
     * @return
     */
    public MessageQueue getMessageQueue(Long netId) {
        return this.clientSessionMap.get(netId);
    }

    public MessageQueue getMessageQueue(ClientConnection clientConnection) {
        return this.clientSessionMap.get(clientConnection.getNetId());
    }

    /**
     * 从分发器中的缓存中删除netId对应的messageQueue
     *
     * @param netId
     */
    public void removeMessageQueue(Long netId) {
        MessageQueue messageQueue = this.clientSessionMap.remove(netId);
        if (messageQueue != null) {
            messageQueue.clear();
        }
    }

    public void removeMessageQueue(ClientConnection clientConnection) {
        MessageQueue messageQueue = this.clientSessionMap.remove(clientConnection.getNetId());
        if (messageQueue != null) {
            messageQueue.clear();
        }
    }

    /**
     * 将消息添加到对应的messageQueue中。
     *
     * @param clientConnection
     * @param message
     */
    public void addMessage(ClientConnection clientConnection, Message message) {

        //创建requestMessage对象
        RequestMessage requestMessage = new RequestMessage(clientConnection, message);
        Long netId = clientConnection.getNetId();
        //从map中查找netId对应的messageQueue
        MessageQueue messageQueue = this.clientSessionMap.get(netId);
        if (messageQueue == null) {
            //不存在, 创建一个新的messageQueue
            messageQueue = new MessageQueue();
            messageQueue.add(requestMessage);
            this.clientSessionMap.put(netId, messageQueue);
        } else {
            messageQueue.add(requestMessage);
        }
        logger.info("向 netId: {} 对应的MessageQueue中添加一个消息 message: {}.", netId,
                message.getClass().getSimpleName());
    }

    @Override
    public void run() {
        // clientMessageDispatcher分发器运行在单独的线程中。
        logger.info("clientMessageDispatcher开始运行...");
        logger.info("分发器正在运行中...");
        while (true) {
            //遍历所有的messageQueue
            for (MessageQueue messageQueue : this.clientSessionMap.values()) {
                //如果有元素
                if (messageQueue != null && messageQueue.size() > 0) {
                    //获得connection
                    //封装成一个Task, 送入线程池进行处理
                    MessageQueueTask messageQueueTask = new MessageQueueTask(messageQueue);
                    this.handleMessageQueueExecutor.execute(messageQueueTask);
                }
            }
            try {
                //轮询的周期
                Thread.sleep(this.sleepTime);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 一个线程任务类。
     * 用于封装一个messageQueue消息队列, 然后交由线程池进行任务的处理。
     *
     * @author tao
     */
    private final class MessageQueueTask implements Runnable {

        private MessageQueue messageQueue;
        private RequestMessage requestMessage;

        //构造函数
        public MessageQueueTask(MessageQueue messageQueue) {
            this.messageQueue = messageQueue;
            /**
             * 注意:
             * 分发器主线程在一次轮询中会为每个非空的messageQueue创建一个对应的messageQueueTask对象。
             * 这个messageQueueTask会从它对应的messageQueue中“只取”一个requestMessage进行处理。
             * 如果messageQueue中有不止一个requestMessage, 会在下一次轮询中再取一个,
             * 这样做的好处是, 对每个客户端都是公平的, 不会因为某个客户的messageQueue中的requestMessage太多,
             * 而阻塞在这个客户端的requestMessage处理上。
             */
            this.requestMessage = this.messageQueue.take();
        }

        /**
         * 取得messageQueue中的一个message之后, 由线程池中的一个线程来处理它。
         */
        @Override
        public void run() {
            try {
                //处理获得的requestMessage
                handleMessage(this.requestMessage);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        /**
         * 处理message
         *
         * @param requestMessage
         */
        private void handleMessage(RequestMessage requestMessage) {

            //获得内部的message对象
            Message message = requestMessage.getMessage();
            logger.info("dispatcher处理一个消息, messageName: [{}]", message.getClass().getSimpleName());

            //根据Message种类, 选择正确的处理方法
            /**
             * 首先需要判断消息是要转发给AuthServer还是LogicServer。
             * 通过ParseMap获取message对应的ptoNum, 然后转发。
             * CRegister、CLogin转发给AuthServer处理;
             * CChatMsg转发给LogicServer处理。
             */
            int ptoNum = ParseMap.getPtoNum(message);

            switch (ptoNum) {

                /**
                 * 如果是客户端的注册、登录消息
                 */
                case PtoNum.CREGISTER:
                case PtoNum.CLOGIN:
                    //发送给AuthServer服务器处理
                    forwardToAuthServer(requestMessage);
                    break;

                /**
                 * 如果是客户端发送的的游戏消息
                 */
                case PtoNum.CGAMEMSG:
                case PtoNum.CGAMEAOI:
                case PtoNum.CGAMEMOVE:
                case PtoNum.CGAMECHATMSG:
                case PtoNum.CGAMESKILLMSG:
                case PtoNum.CGAMEATTACKMSG:
                case PtoNum.CGAMEROLEMSG:
                case PtoNum.CGAMEWEAPONRYMSG:
                case PtoNum.CGAMEITEMMSG:
                case PtoNum.CGAMEUSERLOGININITMSG:
                case PtoNum.GGAMEUSERLOGOUTMSG:
                case PtoNum.CGAMEBOSSSCENEMSG:
                case PtoNum.CGAMESHOPMEG:
                case PtoNum.AUTOFIGHTINGMSG:
                case PtoNum.ONETOONECHAT:
                case PtoNum.ONETOONECHATOLD:
                case PtoNum.WORLDPLAYERCHAT:
                case PtoNum.WORLDPLAYERCHATOLD:
                case PtoNum.CGAMECREATEROLEMSG:
                case PtoNum.CGAMEWORLDPLAYEREMAILMSG:
                case PtoNum.CGAMEPVPSCENEMSG:
                case PtoNum.TEAMMSG:
                case PtoNum.TRADINGMSG:
                case PtoNum.GUILDMSG:
                case PtoNum.TRADINGSHOPMSG:
                case PtoNum.TASKMSG:
                    //发送给logicServer服务器处理
                    forwardToLogicServer(requestMessage);
                    break;

                default:
                    break;
            }

        }

        /**
         * 转发给AuthServer处理.
         *
         * @param requestMessage
         */
        private void forwardToAuthServer(RequestMessage requestMessage) {

            ClientConnection clientConnection = requestMessage.getClientConnection();
            Message message = requestMessage.getMessage();
            ByteBuf sendBuf = null;

            if (message instanceof Auth.CRegister) {
                //如果是注册消息, 先包装成服务器之间的协议体GTransfer
                sendBuf = ServerProtoUtils.pack2Server(Internal.DestType.Auth,
                        clientConnection.getNetId(),
                        ((Auth.CRegister) message).getUserId(),
                        ParseMap.getPtoNum(message), message);
            } else if (message instanceof Auth.CLogin) {
                //在发送CLogin消息之前, 需要先判断是否已经登录
                String userId = ((Auth.CLogin) message).getUserId();
                boolean isLogin = ClientConnectionMap.isUserLogin(userId);
                if (isLogin) {
                    //如果已经登录, 不允许重复登录
                    //向客户端发送重复登录的警告!
                    logger.info("账户已经登录, 不允许重复登录, userId: {}", userId);
                    RouteUtils.sendResponse(Common.REPEATED_LOGIN, "账户已经登入。", clientConnection);
                } else {
                    //如果未登录
                    //登录消息, 先包装成服务器之间的协议体GTransfer
                    sendBuf = ServerProtoUtils.pack2Server(Internal.DestType.Auth,
                            clientConnection.getNetId(),
                            ((Auth.CLogin) message).getUserId(),
                            ParseMap.getPtoNum(message), message);
                }
            }
            if (sendBuf != null) {
                //获取Gate->Auth的连接, 并将消息发送给AuthServer
                logger.info("GateServer 发送请求消息  AuthServer.", message.getClass().getSimpleName());
                ChannelHandlerContext gate2AuthConnCtx = GateToAuthConnectionHandler.getGateToAuthConnectionContext();
                gate2AuthConnCtx.channel().writeAndFlush(sendBuf);
            }
        }

        /**
         * 游戏逻辑转发给LogicServer处理.
         *
         * @param requestMessage
         */
        private void forwardToLogicServer(RequestMessage requestMessage) {
            ClientConnection clientConnection = requestMessage.getClientConnection();
            Message message = requestMessage.getMessage();
            //先判断用户有没有登录
            if (clientConnection.getUserId() == null) {
                //用户Id为空, 说明没有登录就直接发消息了
                logger.info("用户没有登入!");
                RouteUtils.sendResponse(Common.GAME_SENDER_OFFLINE,
                        "请先登入你的用户!", clientConnection);
            } else {
                //客户端发送的逻辑消息,封装成GTransfer发送给logicServer
                ByteBuf sendBuf = ServerProtoUtils.pack2Server(Internal.DestType.Logic,
                        clientConnection.getNetId(), clientConnection.getUserId(),
                        ParseMap.getPtoNum(message), message, clientConnection.getCtx().attr(ClientConnection.SCENEID).get());
                //获得Gate到Logic的连接的ctx, 并将消息发送给LogicServer
                ChannelHandlerContext gate2LogicConnCtx = GateToLogicConnectionHandler.getGateToLogicConnectionContext();
                gate2LogicConnCtx.channel().writeAndFlush(sendBuf);
                logger.info("[GateServer] send [{}] message to [LogicServer]", message.getClass().getSimpleName());
            }
        }
    }
}